/*
**    Copyright (C) 2003-2011 Institute for Systems Biology
**                            Seattle, Washington, USA.
**
**    This library is free software; you can redistribute it and/or
**    modify it under the terms of the GNU Lesser General Public
**    License as published by the Free Software Foundation; either
**    version 2.1 of the License, or (at your option) any later version.
**
**    This library is distributed in the hope that it will be useful,
**    but WITHOUT ANY WARRANTY; without even the implied warranty of
**    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
**    Lesser General Public License for more details.
**
**    You should have received a copy of the GNU Lesser General Public
**    License along with this library; if not, write to the Free Software
**    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*/

package org.systemsbiology.hadoop.ioformats;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import java.io.IOException;
import java.util.List;

/**
 * InputFormat to create [filename, file-line] pairs, as opposed to [line-number, file-line] pairs,
 * to feed to a mapper. Many important analysis algorithms care about the source of a line of
 * data rather than the orders of all lines.
 */
public class FileAwareTextInputFormat extends InputFormat<Text,Text> {
    private TextInputFormat backing;
    public FileAwareTextInputFormat(){
        backing = new TextInputFormat();
    }

    @Override
    public List<InputSplit> getSplits(JobContext jobContext) throws IOException, InterruptedException {
        return backing.getSplits(jobContext);
    }

    private static class FilenameShim extends RecordReader<Text, Text>{

        private RecordReader<?, Text> backingReader;
        private String path;

        public FilenameShim(RecordReader<?, Text> backing){
            backingReader = backing;
        }

        @Override
        public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            try{
                FileSplit asFileSplit = (FileSplit)inputSplit;
                path = asFileSplit.getPath().toString();
            } catch (ClassCastException nope){
                path = "";
            }
            backingReader.initialize(inputSplit, taskAttemptContext);
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            return backingReader.nextKeyValue();
        }

        @Override
        public Text getCurrentKey() throws IOException, InterruptedException {
            return new Text(path);
        }

        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return backingReader.getCurrentValue();
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            return backingReader.getProgress();
        }

        @Override
        public void close() throws IOException {
            backingReader.close();
        }
    }
    @Override
    public RecordReader<Text, Text> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new FilenameShim(backing.createRecordReader(inputSplit, taskAttemptContext));
    }

}
